//package com.generator.listener.zk;
//
//import com.generator.listener.NodeListener;
//import com.generator.listener.event.Event;
//import com.generator.listener.event.EventType;
//import com.generator.service.ZkService;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.curator.framework.recipes.cache.ChildData;
//import org.apache.curator.framework.recipes.cache.NodeCache;
//import org.springframework.beans.BeansException;
//import org.springframework.beans.factory.InitializingBean;
//import org.springframework.context.ApplicationContext;
//import org.springframework.context.ApplicationContextAware;
//import org.springframework.context.annotation.Lazy;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.Resource;
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Map;
//import java.util.concurrent.ArrayBlockingQueue;
//import java.util.concurrent.BlockingQueue;
//import java.util.concurrent.ConcurrentHashMap;
//import java.util.concurrent.LinkedBlockingQueue;
//import java.util.concurrent.ThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//import java.util.concurrent.atomic.AtomicBoolean;
//
//@Slf4j
//@Lazy(false)
//@Service
//public class ZkListenerManager implements InitializingBean, ApplicationContextAware {
//
//    private ApplicationContext applicationContext;
//
//    private final ConcurrentHashMap<String, NodeCache> nodeCacheMap = new ConcurrentHashMap<>(16);
//    private final ConcurrentHashMap<String, List<NodeListener>> nodeListenerMap = new ConcurrentHashMap<>(16);
//
//    private final BlockingQueue<Event> eventQueue = new ArrayBlockingQueue<>(64);
//
//    private ThreadPoolExecutor eventHandlerPool;
//
//    @Resource
//    private ZkService zkService;
//
//    private AtomicBoolean alive = new AtomicBoolean(false);
//
//    @Override
//    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//        this.applicationContext = applicationContext;
//    }
//
//    @Override
//    public void afterPropertiesSet() throws Exception {
//
//        int coreSize = Runtime.getRuntime().availableProcessors() + 1;
//        eventHandlerPool = new ThreadPoolExecutor(coreSize,coreSize << 1,1, TimeUnit.SECONDS,new LinkedBlockingQueue<>());
//        eventHandlerPool.allowCoreThreadTimeOut(true);
//
//        Map<String,NodeListener> beanMap = applicationContext.getBeansOfType(NodeListener.class);
//
//        for (Map.Entry<String,NodeListener> entry : beanMap.entrySet()) {
//
//            NodeListener nodeListener = entry.getValue();
//            List<String> listenerPaths = nodeListener.getPath();
//
//            for (String listenerPath : listenerPaths) {
//
//                log.info("[ZkEventManager] 注册事件监听器 >>> path:{}",listenerPath);
//                if(!nodeListenerMap.containsKey(listenerPath)) {
//                    nodeListenerMap.putIfAbsent(listenerPath,new ArrayList<>(4));
//                }
//
//                List<NodeListener> listenerList = nodeListenerMap.get(listenerPath);
//                listenerList.add(nodeListener);
//
//                if(nodeCacheMap.get(listenerPath) == null) {
//                    NodeCache nc = new NodeCache(zkService.getClient(),listenerPath);
//                    nodeCacheMap.putIfAbsent(listenerPath,nc);
//
//                    nc.getListenable().addListener(() -> {
//
//                        Event event;
//
//                        ChildData childData = nc.getCurrentData();
//
//                        if(childData == null || childData.getStat() == null) {
//                            event = Event.builder()
//                                    .eventType(EventType.NODE_DELETE)
//                                    .path(listenerPath)
//                                    .data(null)
//                                    .build();
//                        } else {
//                            event = Event.builder()
//                                    .eventType(EventType.NODE_DATA_CHANGE)
//                                    .path(childData.getPath())
//                                    .data(new String(childData.getData(),ZkService.DEFAULT_CHARSET))
//                                    .build();
//                        }
//
//                        eventQueue.put(event);
//
//                    });
//                }
//
//            }
//        }
//
//        for (Map.Entry<String,NodeCache> entry : nodeCacheMap.entrySet()) {
//            entry.getValue().start();
//        }
//
//        startEventHandleThread();
//
//        registerHook();
//    }
//
//    private void startEventHandleThread() {
//
//        Thread thread = new Thread(() -> {
//
//            alive.set(true);
//
//            while (alive.get()) {
//
//                try {
//
//                    Event event = eventQueue.take();
//
//                    log.info("[EventHandleThread] 处理事件 >>> eventType:{},path:{}, data:{}",event.getEventType().name(),event.getPath(),event.getData());
//
//                    List<NodeListener> nodeListenerList = nodeListenerMap.get(event.getPath());
//
//                    for (NodeListener nodeListener : nodeListenerList) {
//
//                        if(!alive.get()) {
//                            break;
//                        }
//
//                        eventHandlerPool.execute(() -> {
//
//                            try {
//                                if(event.getEventType() == EventType.NODE_DELETE) {
//                                    nodeListener.onDelete(event.getPath());
//                                } else if(event.getEventType() == EventType.NODE_DATA_CHANGE) {
//                                    nodeListener.onDataChange(event.getPath(),event.getData());
//                                }
//                            } catch (Exception e) {
//                                // ignore
//                            }
//
//                        });
//                    }
//
//                } catch (Exception e) {
//                    // ignore
//                }
//            }
//
//        },"[ZkListenerManager-EventHandleThread]");
//
//        thread.setDaemon(true);
//        thread.start();
//
//
//    }
//
//    private void registerHook() {
//        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
//
//            alive.set(false);
//
//            eventHandlerPool.shutdown();
//
//        },"ID-SERVICE-HOOK-THREAD"));
//    }
//}
